package com.wechat.order.platform.webScoket;

import com.wechat.order.platform.redisConverAndSend.RedisHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.security.crypto.codec.Utf8;
import org.springframework.stereotype.Component;

import javax.websocket.OnClose;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * Created by zzy on 2017/12/21.
 */
@Component
@ServerEndpoint("/webSocket")
@Slf4j
public class WebSocket {

    private Session session;

    private final ConcurrentMap<String, RedisLisener> redisListenerMap = new ConcurrentHashMap<>();

    @Autowired
    private  RedisMessageListenerContainer redisMessageListenerContainer;

    private static CopyOnWriteArrayList<WebSocket> webSocketSet=new CopyOnWriteArrayList<>();



    @OnOpen
    public void onOpen(Session session){
        this.session=session;
        webSocketSet.add(this);
        log.info("[webSocket消息] 有新的链接，总数：{}",webSocketSet.size());
      //  addRedisMessageListenerContainer(this);

    }

    @OnClose
    public void onClose(Session session){
        this.session=session;
        webSocketSet.remove(this);
        log.info("[webSocket消息] 有链接断开，总数：{}",webSocketSet.size());
    }

    @OnMessage
    public void onMessage(String message){
        log.info("[websocket消息] 收到客户端发来的消息：{}",message);
    }

    public void sendMessage(String message){
        for (WebSocket webSocket:webSocketSet){
            log.info("[websocket消息] 广播消息，message",message);
            try {
                webSocket.session.getBasicRemote().sendText(message);
            } catch (IOException e) {
                log.info("[websocket消息] 广播消息失败" +e.getMessage());
            }
        }
    }

    /**
     * 获取当前连接session
     * @return
     */
    public Session getSession(){
        return session;
    }



    /**
     *添加redis发布订阅
     */
    private void addRedisMessageListenerContainer(WebSocket webSocket){
        redisListenerMap.put(webSocket.getSession().getId().toString(), new RedisLisener(webSocket));
        RedisLisener redisLisener = redisListenerMap.get(webSocket.getSession().getId().toString());
        redisMessageListenerContainer
                .addMessageListener(redisLisener, new ChannelTopic("hello" ));
    }


    public static class RedisLisener implements MessageListener {

        private final WebSocket webSocket;

        RedisLisener(WebSocket socket) {
            this.webSocket = socket;
        }
        @Override
        public void onMessage(Message message, byte[] bytes) {
            String json= Utf8.decode(message.getBody());
            webSocket.sendMessage(json);
        }
    }
}
